import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.core.TransactionalMap;
import com.hazelcast.transaction.TransactionContext;

/**
 * transaction map 开始后，读取相关要素的线程将被阻塞
 * 
 * @author wxu
 *
 */
public class Transactional {

	private static final Logger logger = LoggerFactory.getLogger(Transactional.class);
	private static AtomicBoolean flag = new AtomicBoolean(false);

	private static IMap<Object, Object> testmap;

	static Runnable tx = new Runnable() {
		@Override
		public void run() {

			try {
				logger.info("start transaction...");
				TransactionContext txCxt = hz.newTransactionContext();
				txCxt.beginTransaction();
				TransactionalMap<Object, Object> map = txCxt.getMap("map");

				try {
					logger.info("before put map(1)");
					Thread.sleep(300);
					map.put("1", "1"); // reader1 is blocked
					logger.info("after put map(1)");
					Thread.sleep(500);
					map.put("2", "2"); // reader2 is blocked
					logger.info("after put map(2)");
					Thread.sleep(500);
					txCxt.commitTransaction();
					logger.info("transaction committed");
				} catch (RuntimeException t) {
					txCxt.rollbackTransaction();
					throw t;
				}

				Thread.sleep(500);
			} catch (InterruptedException e) {
				e.printStackTrace();
			} finally {
				logger.info("Finished testmap size:{}, testmap(1):{}, testmap(2):{} ", testmap.size(), testmap.get("1"),
						testmap.get("2"));

				Hazelcast.shutdownAll();
				logger.info("system exit.");
				System.exit(0);
			}

		}
	};

	static Runnable check = new Runnable() {
		public void run() {
			// check map size
			while (true) {
				logger.info("testmap size:{}, testmap(1):{}, testmap(2):{} ", testmap.size(), testmap.get("1"),
						testmap.get("2"));
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	};
	static Runnable reader1 = new Runnable() {
		public void run() {
			// check map size
			while (true) {
				logger.info("testmap(1):{}", testmap.get("1"));
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	};
	static Runnable reader2 = new Runnable() {
		public void run() {
			// check map size
			while (true) {
				logger.info("testmap(2):{}", testmap.get("2"));
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	};
	static Runnable reader3 = new Runnable() {
		public void run() {
			// check map size
			while (true) {
				logger.info("testmap(3):{}", testmap.get("3"));
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	};

	private static HazelcastInstance hz;

	public static void main(String[] args) throws Exception {
		hz = Hazelcast.newHazelcastInstance();

		testmap = hz.getMap("map");
		testmap.set("3", "3");
		// new Thread(check).start();
		new Thread(reader1, "reader1").start();
		new Thread(reader2, "reader2").start();
		new Thread(reader3, "reader3").start();

		Thread.sleep(500);
		new Thread(tx).start();
		flag.set(true);

		// Hazelcast.shutdownAll();
		// System.exit(0);
	}
}
